package progress.exa28_2.part1;

import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConsumerProducer {

    private static Buffer buffer = new Buffer();

    public static void main(String[] args) {
        ExecutorService executor= Executors.newFixedThreadPool(10);//创建一个并行线程池
        executor.execute(new ProducerTask());
        executor.execute(new ConsumerTask());
        executor.execute(new ConsumerTask());
        executor.execute(new ConsumerTask());
        executor.execute(new ConsumerTask());
        executor.execute(new ConsumerTask());
        executor.shutdown();
    }

    private static class ProducerTask implements Runnable{
        public void run(){
            try{
                int i=1;
                while(true){
                    System.out.println("Producer writes"+i);
                    buffer.write(i++);
                    Thread.sleep((int)(Math.random()*1000));//线程休眠
                }
            }catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }

        }
    }

    private static class ConsumerTask implements Runnable {
        public void run() {
            try {
                while (true) {
                    System.out.println("\t\t\tConsumer reads"+buffer.read());
                    Thread.sleep((int)(Math.random()*1000));//线程休眠
                }
            }catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }

        }
    }

    /**
     * 线程安全的缓冲类
     */
    private static class Buffer {

        private static final int CAPACITY = 10;//缓冲区大小为10

        private final LinkedList<Integer> queue = new LinkedList<>();

        //创建互斥锁
        private static final Lock lock = new ReentrantLock();

        //创建条件对象，用于线程间通信
        private static final Condition notEmpty = lock.newCondition();
        private static final Condition notFull = lock.newCondition();

        public void write(int value) {
            //让当前线程获得锁
            lock.lock();
            try{
                //检查队列是否满了
                if(queue.size()==CAPACITY){
                    System.out.println("缓冲区满了，等待读取");
                    //释放锁，线程进入等待状态
                    notFull.await();
                }
                //添加value到queue
                queue.offer(value);
                //唤醒阻塞的读线程
                notEmpty.signal();
            }catch(InterruptedException ex) {
                ex.printStackTrace();
            }
            finally{
                //当前线程释放锁
                lock.unlock();
            }
        }

        public int read() {
            int value=0;
            //让当前线程获得锁
            lock.lock();
            try{
                //检查队列是否空了
                if(queue.isEmpty()){
                    System.out.println("\t\t\t缓冲区空了，等待写入");
                    //释放锁，当前线程进入等待状态
                    notEmpty.await();
                }
                //取出队列的元素
                value = queue.remove();
                //唤醒写线程
                notFull.signal();
            }catch(InterruptedException ex){
                ex.printStackTrace();
            } finally {
                //当前线程释放锁
                lock.unlock();
            }
            return value;
        }

    }
}